package org.study.hadoop.md.ch2;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.study.hadoop.md.util.MRDPUtils;
/**
 * 计数器技数1：使用Hadoop用户技术器来统计每个州的用户数。
 */
public class CountNumUsersByStateDriver extends Configured implements Tool {
	
	public static class CountNumUsersByStateMapper extends Mapper<Object, Text, NullWritable, NullWritable> {
		public static final String STATE_COUNTER_GROUP = "State";
		private String[] statesArray = new String[] { "AL", "AK", "AZ", "AR",
				"CA", "CO", "CT", "DE", "FL", "GA", "HI", "ID", "IL", "IN",
				"IA", "KS", "KY", "LA", "ME", "MD", "MA", "MI", "MN", "MS",
				"MO", "MT", "NE", "NV", "NH", "NJ", "NM", "NY", "NC", "ND",
				"OH", "OK", "OR", "PA", "RI", "SC", "SF", "TN", "TX", "UT",
				"VT", "VA", "WA", "WV", "WI", "WY" };
		private HashSet<String> states = new HashSet<String>(Arrays.asList(statesArray));
		
		@Override
		protected void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			Map<String, String> parsed = MRDPUtils.transformXmlToMap(value.toString());
			String location = parsed.get("Location");
			if (location != null && !location.isEmpty()) {
				boolean unknow = true;
				String[] tokens = location.toUpperCase().split("-");
				for (String state : tokens) {
					if (states.contains(state)) {
						context.getCounter(STATE_COUNTER_GROUP, state).increment(1);
						unknow = false;
						break;
					}
				}
				if (unknow) {
					context.getCounter(STATE_COUNTER_GROUP, "Unknown").increment(1);
				}
			} else {
				context.getCounter(STATE_COUNTER_GROUP, "NullOrEmpty").increment(1);
			}
		}
	}
	
	@Override
	public int run(String[] args) throws Exception {
		Path outPath = new Path("out");
		Path inPath = new Path("user.txt");
		FileSystem fs = outPath.getFileSystem(getConf());
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}
		Job job = Job.getInstance(getConf());
		job.setJarByClass(CountNumUsersByStateDriver.class);
		job.setMapperClass(CountNumUsersByStateMapper.class);
		job.setNumReduceTasks(0);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(NullWritable.class);
		FileInputFormat.addInputPath(job, inPath);
		FileOutputFormat.setOutputPath(job, outPath);
		return job.waitForCompletion(true) ? 0 : 1;
	}
	
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new CountNumUsersByStateDriver(), args);
		System.exit(res);
	}
}